package com.mjf.app.dws;

import com.alibaba.fastjson.JSON;
import com.mjf.bean.OrderWide;
import com.mjf.bean.ProvinceStats;
import com.mjf.utils.ClickHouseUtil;
import com.mjf.utils.DateTimeUtil;
import com.mjf.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;

/**
 * 地区主题宽表计算
 * <p>
 * 数据流向：web/app -> nginx -> springboot -> MySQL -> flinkApp -> kafka(ods) -> flinkApp -> kafka(dwd)/HBase(dim)
 * -> flinkApp(redis) -> Kafka(dwm) -> flinkApp -> clickhouse
 * 程序：mockDb(包含web/nginx/springboot) -> MySQL -> FlinkCDC(包含flinkApp/kafka(ods)) -> BaseDbApp(包含flinkApp/kafka(dwd)/HBase(dim))
 * -> OrderWideApp(包含flinkApp/Kafka(dwm)) -> ProvinceStatsApp(包含flinkApp/clickhouse(dws))
 * 环境：hdfs zookeeper kafka hbase phoenix(bin/sqlline.py) redis clickhouse
 */
public class ProvinceStatsApp {
    public static void main(String[] args) throws Exception {

        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.1 设置CK&状态后端
        //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall-flink-210325/ck"));
        //env.enableCheckpointing(5000L);
        //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);

        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart());

        // 2.提取时间戳生成WaterMark
        String groupId = "ProvinceStatsSqlApp";
        String orderWideTopic = "dwm_order_wide";

        SingleOutputStreamOperator<ProvinceStats> provinceStatsDS = env
                .addSource(MyKafkaUtil.getKafkaConsumer(orderWideTopic, groupId))
                .map(line -> JSON.parseObject(line, OrderWide.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderWide>() {
                                    @Override
                                    public long extractTimestamp(OrderWide element, long recordTimestamp) {
                                        return DateTimeUtil.toTs(element.getCreate_time());
                                    }
                                })
                )
                .map(ProvinceStats::new);

        // 3.分组、开窗、聚合
        SingleOutputStreamOperator<ProvinceStats> resultDS = provinceStatsDS
                .keyBy(new KeySelector<ProvinceStats, Tuple5<Long, String, String, String, String>>() {
                    @Override
                    public Tuple5 getKey(ProvinceStats provinceStats) throws Exception {
                        return new Tuple5<Long, String, String, String, String>(
                                provinceStats.getProvince_id(),
                                provinceStats.getProvince_name(),
                                provinceStats.getProvince_area_code(),
                                provinceStats.getProvince_iso_code(),
                                provinceStats.getProvince_3166_2_code()
                        );
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .apply(new WindowFunction<ProvinceStats, ProvinceStats, Tuple5<Long, String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple5<Long, String, String, String, String> key, TimeWindow window, Iterable<ProvinceStats> input, Collector<ProvinceStats> out) throws Exception {
                        // 生成迭代器
                        Iterator<ProvinceStats> iterator = input.iterator();

                        // 创建对象用于返回
                        ProvinceStats provinceStats = new ProvinceStats();
                        provinceStats.setStt(DateTimeUtil.toYMDhms(new Date(window.getStart())));
                        provinceStats.setEdt(DateTimeUtil.toYMDhms(new Date(window.getEnd())));
                        provinceStats.setProvince_id(key.getField(0));
                        provinceStats.setProvince_name(key.getField(1));
                        provinceStats.setProvince_area_code(key.getField(2));
                        provinceStats.setProvince_iso_code(key.getField(3));
                        provinceStats.setProvince_3166_2_code(key.getField(4));
                        provinceStats.setTs(new Date().getTime());

                        // 用于订单id去重
                        HashSet<Long> orderIds = new HashSet<>();

                        // 遍历分区数据
                        while (iterator.hasNext()) {
                            ProvinceStats next = iterator.next();
                            BigDecimal order_amount = next.getOrder_amount();
                            Long order_id = next.getOrder_id();

                            if (provinceStats.getOrder_amount() == null) {
                                provinceStats.setOrder_amount(order_amount);
                            } else {
                                provinceStats.setOrder_amount(provinceStats.getOrder_amount().add(order_amount));
                            }

                            orderIds.add(order_id);
                        }

                        provinceStats.setOrder_count((long) orderIds.size());

                        // 返回结果
                        out.collect(provinceStats);
                    }
                });

        // 4.打印数据并写入ClickHouse
        resultDS.print();
        resultDS.addSink(ClickHouseUtil.getSink(
                "insert into province_stats values(?,?,?,?,?,?,?,?,?,?)"
        ));

        // 6.启动任务
        env.execute("ProvinceStatsApp");

    }
}
